package com.game.logic.service;

import com.game.logic.messages.GameMessageServerBoot;
import com.game.logic.channel.GameServerContext;
import com.game.logic.netty.RequestHandlerContext;
import com.game.logic.netty.HandlerMethod;
import com.game.framework.message.event.IGameMessage;
import com.game.framework.message.stream.AbstractMessageHeader;
import com.game.framework.message.stream.ServiceMessageHeader;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

@Slf4j
@Service
@RocketMQMessageListener(topic = "${mq.topic}-${game.common.config.local-server-id}", consumerGroup = "${mq.consumer.group}", consumeMode = ConsumeMode.ORDERLY
)
//监听网关的消息。
public class ReceiveConsumeMessageService implements RocketMQListener<MessageExt> {

    @Resource
    protected RequestHandlerContext requestHandlerContext;

    @Autowired
    protected Environment environment;

    private String topic;

    @PostConstruct
    public void init() {
        RocketMQMessageListener rocketMQMessageListener = this.getClass().getAnnotation(RocketMQMessageListener.class);
        if(rocketMQMessageListener == null) {
            log.error("没有添加消费者消息监听");
            return ;
        }
        topic = environment.resolveRequiredPlaceholders(rocketMQMessageListener.topic());
        String consumerGroup = environment.resolveRequiredPlaceholders(rocketMQMessageListener.consumerGroup());
        log.info("监听消费消息 topic:{}, group:{}", topic, consumerGroup);
    }
    
    public String getTopic() {
        return topic;
    }


    @Override
    public void onMessage(MessageExt message) {
        if (!GameMessageServerBoot.getIsStarted()) {
            log.warn("服务未启动成功，收到消息暂不处理");
            return;
        }
        IGameMessage gameMessage;
        try {
            gameMessage = this.getGameMessage(message.getBody());
            if (gameMessage == null) {
                return;
            }
            ServiceMessageHeader header = (ServiceMessageHeader) gameMessage.getHeader();
            HandlerMethod handlerMethod = this.requestHandlerContext.getHandlerMethod(header.getMessageId());
            GameMessageServerBoot.messageServerContext.getServerChannelGroup().gameChannelConsumer(header.getUserId(), gameChannel -> {
                gameChannel.submitTask(() -> {
                    try {
                        log.debug("=>{}", header);
                        GameServerContext gameServerContext = new GameServerContext(header, gameChannel);
                        handlerMethod.getTargetMethod().invoke(handlerMethod.getTargetObj(), gameMessage, gameServerContext);
                    } catch (Throwable e) {
                        log.error("消费者消息处理出错", e);
                    }
                }, "处理消费者消息");
            });

        } catch (Exception e) {
            log.error("解析消费者数据出错", e);
        }
    }

    private IGameMessage getGameMessage(byte[] data) throws Exception {
        ByteBuf byteBuf = Unpooled.wrappedBuffer(data);
		int messageId = AbstractMessageHeader.readMessageId(byteBuf);
        IGameMessage gameMessage = requestHandlerContext.getRequestMessage(messageId);
        if(gameMessage != null) {
        	gameMessage.read(byteBuf);
        } else {
        	log.error("找不到messageId {} 对应的消息对象",messageId);
        }
        return gameMessage;
    }

    public RequestHandlerContext getRequestHandlerContext() {
        return requestHandlerContext;
    }

    public void setRequestHandlerContext(RequestHandlerContext requestHandlerContext) {
        this.requestHandlerContext = requestHandlerContext;
    }

    public Environment getEnvironment() {
        return environment;
    }

    public void setEnvironment(Environment environment) {
        this.environment = environment;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }
}
